package com.diego.strom.demo.wordcount;

import java.util.Arrays;

import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.StormSubmitter;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.tuple.Fields;

/**
 * 将代码通过eclipse export 导出包为：WordCount.jar， 上传到storm集群中，执行命令如下：
 *  启动：./storm jar wordcount.jar com.diego.strom.demo.wordcount.WordCountTopologyMain 
 *  停止：storm kill WordCountTopologyMain
 * @author Manfred
 *
 */
public class WordCountTopologyMain {
	
	public static final String NIMBUS_HOST="10.10.10.10";
	
	public static final int NIMBUS_THRIFT_PORT=6627;
	
	public static final String STORM_ZOOKEEPER_SERVERS="10.10.10.10";
	
	public static final int STORM_ZOOKEEPER_PORT=2181;
	
	public static void main(String[] args) throws InterruptedException {
		// 定义一个Topology
		TopologyBuilder builder = new TopologyBuilder();
		builder.setSpout("word-reader", new WordRichSpout());
		builder.setBolt("word-normalizer", new WordNormalizerRichBolt()).shuffleGrouping("word-reader");
		builder.setBolt("word-counter", new WordCounterRichBolt(), 2).fieldsGrouping("word-normalizer",new Fields("word"));
		// 配置
		Config conf = new Config();
		conf.put("wordsFile", "/root/text.txt");
		conf.setDebug(false);
			  
		  
		// 创建一个本地模式cluster
		LocalCluster cluster = new LocalCluster();
		cluster.submitTopology("Getting-Started-Toplogie", conf,builder.createTopology()); 
		Thread.sleep(1000); 
		cluster.shutdown();
		
		// 远程提交Topology
		/*conf.put(Config.TOPOLOGY_MAX_SPOUT_PENDING, 1);
		conf.put(Config.NIMBUS_HOST, NIMBUS_HOST); // 配置nimbus连接主机地址，比如：192.168.10.1
		conf.put(Config.NIMBUS_THRIFT_PORT, NIMBUS_THRIFT_PORT);// 配置nimbus连接端口，默认
																// 6627
		conf.put(Config.STORM_ZOOKEEPER_SERVERS, Arrays.asList(STORM_ZOOKEEPER_SERVERS)); // 配置zookeeper连接主机地址，可以使用集合存放多个
		conf.put(Config.STORM_ZOOKEEPER_PORT, STORM_ZOOKEEPER_PORT); // 配置zookeeper连接端口，默认2181
		conf.setDebug(true);
		conf.setNumWorkers(3); */ 
		//非常关键的一步，使用StormSubmitter提交拓扑时，不管怎么样，都是需要将所需的jar提交到nimbus上去，如果不指定jar文件路径，  
		//storm默认会使用System.getProperty("storm.jar")去取，如果不设定，就不能提交  
		//System.setProperty("storm.jar","d:\\storm-remote-submit-1.0-SNAPSHOT-jar-with-dependencies.jar");  
		//StormSubmitter.submitTopology(name, conf, topology);  
		

		/*StormSubmitter stormSubmitter = new StormSubmitter();// storm集群执行
		try {
			stormSubmitter.submitTopology(WordCountTopologyMain.class.getSimpleName(), conf, builder.createTopology());
		} catch (AlreadyAliveException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (InvalidTopologyException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (AuthorizationException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}*/
	}
}
